package api;
import java.util.Arrays;
import java.util.Iterator;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

public class KafkaConsumerDemo {
    public static void main(String[] args) {
		Properties props=new Properties();
		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.196.128:9092,192.168.196.128:9093,192.168.196.128:9094");
		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ObjectDeSerializer.class);
		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ObjectDeSerializer.class);
		
		props.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");
		
		KafkaConsumer<String, Order> kafkaConsumer=new KafkaConsumer<String, Order>(props);
		kafkaConsumer.subscribe(Arrays.asList("topic01"));
		System.out.println("group1 A...");
		while(true){
			ConsumerRecords<String, Order> records = kafkaConsumer.poll(2000);
			Iterator<ConsumerRecord<String, Order>> iterator = records.iterator();
			while(iterator.hasNext()){
				ConsumerRecord<String, Order> record = iterator.next();
				System.out.println(record.key()+" ,"+record.value()+", "+record.timestamp()+", "+record.partition()+" ,"+record.offset());
			}
		}
		    	
	}
}
